 /********************************************************************
  Copyright by TAOBAO 2009-2010. All rights reserved
  File Name: main.cpp
  Created Date: 2009-06-04
  Author: ALI
  Version: 0.0.1
  Last Version Date:
  Version Date:
  Description: main body for dbloader
  Modified history:
  ver   Date        Author    Description
--------------------------------------------------------------------
0.0.1   20090604 ALI        Created
0.0.2   20100405 ALI        Add hdfs support, dtfmt option, file glob function
                            Fix hdfs read block boundary bug
0.0.3   20100425 ALI        Fix return code when error over limit
                            Add auto guess field and cols feature
0.0.4   20100712 ALI        Check the parameters
                            Check the hdfs path
********************************************************************/

//Connect to Database By ODBC
//#include <sql.h>
//#include <sqlext.h>

#include "dbloader.h"   
#include "strsplit.h"

#define	DBLOADER_VERSION (const char*)"0.0.4"

//map<string, Create_t *, less<string> > factory;		// proxy class

static void PrintUsage(string strPrgName, int ret);


const char Hex2Char( const string strHex )
{
    int nNum;
    if( 0 == strHex.length() )
        return 0;
    if( 1 == strHex.length() )
        return strHex[0];
    sscanf( strHex.c_str(), "0x%x", &nNum);
    return nNum;
}


int main(int argc,char *argv[])
{
    string strPrgName = basename( argv[0] );

    map<string, string> mOptions;
    char chField  = '\0';
    char chRecord = '\0';
    bool bCanEmpty = false;

    // default parameters
    mOptions["user"]   = "";              // must supply
    mOptions["table"]  = "";              // must supply
    mOptions["file"]   = "tabledata.txt"; // default file

    mOptions["field"]  = "";             // field seperator
    mOptions["limit"]  = "0";             // default no errors
    mOptions["skip"]   = "0";             // skip line count
    mOptions["mode"]   = "append";        // append|truncate}truncpart
    mOptions["part"]   = "";              // when using truncpart mode
    mOptions["log"]    = "";

    mOptions["ini"]    = getenv("HOME");
    mOptions["ini"]    += "/.dbloader.ini";

    mOptions["pre"]    = "";              // SQL before load
    mOptions["post"]   = "";              // SQL after  load

    mOptions["dtfmt"]  = "";              // Date Format
    mOptions["canempty"]  = "false";           // can empty?
    mOptions["encode"]  = "";             // Encode Type
    mOptions["order"]  = "";              // ASCII or column list(sep by ',')
    mOptions["parallel"] = "0";           // default not parallel

    try {
    // receive parameters from commandline
    map<string, string>::iterator miter;
    for( int i = 1; i < argc; i++)
    {
        bool bFind = false;
        for( miter = mOptions.begin(); miter != mOptions.end(); miter++)
        {
            string strOpt = miter->first;
            if( 0 == strncasecmp(strOpt.c_str(), argv[i], strOpt.length()))
            {
                mOptions[strOpt] = argv[i] + strOpt.length() + 1;
                bFind = true;
                break;
            }
        }
        if( !bFind )
        {
            cerr << "Error:Parameter[" << argv[i] << "] not support!" << endl;
            exit(1);
        }
    }

    // check and convert parameters
    if( 0 == mOptions["table"].length() )
        PrintUsage( strPrgName, 1);

    chField  = Hex2Char( mOptions["field"]  );
    mOptions["field"]  = chField;

    if(mOptions["mode"] == "truncpart" && mOptions["part"].length() == 0 )
    {
         cerr << "Error:Must have part when using truncpart MODE!" << endl;
         PrintUsage( strPrgName, 1);
    }

    if( 0 == strncasecmp(mOptions["canempty"].c_str(),"false",5) )
        bCanEmpty = false;
    else if( 0 == strncasecmp(mOptions["canempty"].c_str(),"true",4) )
        bCanEmpty = true;
    else
    {
         cerr << "Error:Option[canempty] must be true or false!" << endl;
         PrintUsage( strPrgName, 1);
    }

    if( NULL != getenv("HADOOP_CONF_DIR"))
        int nRet = chdir(getenv("HADOOP_CONF_DIR"));

    // DEBUG
    for( miter = mOptions.begin(); miter != mOptions.end(); miter++)
        cout << miter->first << ":\t\t" << miter->second << endl;

    // load dynamic library according to dbtype
    map<string, string> mLib;
    mLib["db2"] = "libdb2ldr.so";
    mLib["oracle"] = "liboraldr.so";
    mLib["mysql"] = "libmsqlldr.so";

    // argument dbtype has high priority,
    // so you can run without inifile totally
    string strDBType;
    if( mOptions["dbtype"].length() == 0 )
    {
        CIniFile iniFile( mOptions["ini"] );
        if( !iniFile.Read())
        {
            cerr << "Read ini file fail:" << iniFile.GetErrMsg() << endl;
            return -1;
        }
        iniFile.GetValue("load", "dbtype", strDBType);
    }
    else
        strDBType = mOptions["dbtype"];

    for(string::iterator iter = strDBType.begin(); iter != strDBType.end(); iter++)
        *iter = tolower(*iter);
    if(mLib.find(strDBType) == mLib.end())
    {
        cerr << "Unknow Database Type:[" << strDBType << "]" << endl;
        return -1;
    }

    string strLibPath, strLib;
    if( NULL != getenv("LD_LIBRARY_PATH") )
        strLibPath = getenv("LD_LIBRARY_PATH");      // system lib path
    else
        strLibPath = "/lib:/usr/lib:/usr/local/lib:lib"; // standard lib path

    CStrSplit spt;
    spt.Split(strLibPath.c_str(), ":");
    int i = 1;
    for( ; i < spt.GetFieldCount()+1; i++)
    {
        strLib = string(spt[i]) + "/" + mLib[strDBType];
        if( 0 == access( strLib.c_str(), R_OK))
            break;
    }
    if( i == spt.GetFieldCount()+1 )
    {
        cerr << "Error:Not Found LIBRARY[" << mLib[strDBType ]<< "]!" << endl;
        return -1;
    }

    cout << "----------------------------------------------------------" << endl;
    cout << "Library is " << strLib << endl;
    void *dlres = dlopen(strLib.c_str(), RTLD_NOW);
    if(dlres == NULL)
    {
        cerr << "LIBRARY[" << strLib << "]:" << dlerror() << endl;
        return -1;
    }

    // reset dlerror flag
    dlerror();

    Create_t *Create = (Create_t *)dlsym(dlres, "Create");
    DBLoader *clLoadObj = Create();

    clLoadObj->SetLogon( mOptions["user"] );
    clLoadObj->SetIniFile( mOptions["ini"] );
    clLoadObj->SetTable( mOptions["table"] );
    clLoadObj->SetPreSQL( mOptions["pre"] );
    clLoadObj->SetPostSQL( mOptions["post"] );
//    clLoadObj->SetMsgFile(argv[3]);
//    clLoadObj->SetLogFile(argv[3]);
//    clLoadObj->SetDumpFile(argv[3]);
    clLoadObj->SetSepChar( chField );
//    clLoadObj->SetRecChar(argv[3]);
//    clLoadObj->SetEnClosed(argv[3]);
    clLoadObj->SetDateFormat( mOptions["dtfmt"] );
    clLoadObj->SetEncodeType( mOptions["encode"] );
    clLoadObj->SetColOrder( mOptions["order"] );
    clLoadObj->SetLimit( atof(mOptions["limit"].c_str()) );

    if(0 == strncasecmp(mOptions["mode"].c_str(), "append", 6))
        clLoadObj->SetLoadMode( DBLoader::APPEND );
    else if(0 == strncasecmp(mOptions["mode"].c_str(), "truncate", 8))
        clLoadObj->SetLoadMode( DBLoader::TRUNCATE );
    else if(0 == strncasecmp(mOptions["mode"].c_str(), "truncpart", 9))
        clLoadObj->SetLoadMode( DBLoader::TRUNCPART, mOptions["part"] );
    else if(mOptions["mode"].length() > 0)
    {
        cerr << "Error: Unknown load mode[" << mOptions["mode"] << "]!" << endl;
        PrintUsage( strPrgName, 1);
    }

    // generate filelist
    CIFile file;
    vector<string> vFiles = file.Glob( mOptions["file"] );
    if( vFiles.size() == 0 )
    {
        cerr << "Error:File[" << mOptions["file"] << "] can't be found!" << endl;
        exit(-1);
    }

    for( vector<string>::iterator iter = vFiles.begin(); iter != vFiles.end(); iter++)
    {
        cout << "Add File:" << *iter << endl;
        clLoadObj->AddFile( *iter );
    }


    int pLoadResult[6] = {0};
    bool bChkOK = false;
    try {
        clLoadObj->LoaderInit();      // envinit, connect, allocate handles
        clLoadObj->PreLoad();         // execute preSQL, truncate and so on
        clLoadObj->RunLoad();
        bChkOK = clLoadObj->QCCheck();
        clLoadObj->CommitLoad( bChkOK );
        if( bChkOK )                      // when qc check pass, then execute post sql
            clLoadObj->PostLoad();        // execute postSQL, run statistics and so on
        clLoadObj->GetLoadResult(pLoadResult);
    }
    catch(CException &e)
    {
        cerr << "Error:" << e.GetErrMsg() << endl;
        return -1;
    }

    int nPid = getpid();
    int nTotalRec = pLoadResult[0];
    int nSkipRec  = pLoadResult[1];
    int nFailRec  = pLoadResult[1] + pLoadResult[3] + pLoadResult[4];
    int nSuccRec  = pLoadResult[2];
    int nDelRec   = pLoadResult[3];
    int nRejRec   = pLoadResult[4];
    int nLoadRec  = pLoadResult[5];
    int nLoadRet  = pLoadResult[6];

    cout << "Summary:" << endl;
    cout << "  Files:" << vFiles.size()    << endl;
    cout << "  Total:" << nTotalRec << endl;
    cout << " Loaded:" << nSuccRec  << endl;
    cout << " Failed:" << nFailRec  << endl;

    return !bChkOK;

    } catch(CException &e)
    {
        cerr << "Error:" << e.GetErrMsg() << endl;
        return -1;
    }
}


void PrintUsage(string strPrgName, int ret)
{
    printf("Usage: %s user=... table=... file=... field=... limit=...\n", strPrgName.c_str());
    printf("(@) Copyright Wang Yong 2009, all rights reserved.\n");
    printf("Notes:\n");
    printf("       -si   = enable logon as SYSDBA\n");
    printf("       user  = username/password@tnsname\n");
    printf("       table = table name the data will loaded into\n");
    printf("       file  = input files name(c-pattern)\n");
    printf("       order = column order,[ascii|col1,col2,col3...]\n");
    printf("       limit = percent when less than 1, rows when large than 1\n");
    printf("       field = seperator string between fields\n");
    printf("       mode  = load mode, [append|truncate|truncpart]\n");
    printf("       part  = partition list, using with truncpart mode\n");
    printf("       dtfmt = date format, using with date columns[YYYYMMDD]\n");
    printf("      encode = encoding type, should be one of [GBK|UTF8]\n");
    printf("       ini   = ini file name, default use ~/.dbload.ini\n");
    printf("       log   = log file name, prefix with + to append mode\n");
    printf("   canempty  = if file can be empty, [true|false],default false\n");
    printf("    parallel = count of parallel file\n");
    printf("\n");
    printf("         pre = the sql executed before load\n");
    printf("        post = the sql executed after load\n");
//    printf("       long  = maximum long field size\n");                        
//    printf("       array = array fetch size\n");                               
//    printf("       buffer= sqlldr READSIZE and BINDSIZE, default 16 (MB)\n");  
    printf("\n");                                                              
    printf("  for field, you can use '0x' to specify hex character code,\n");     
    printf("  \\r=0x%02x \\n=0x%02x |=0x%0x ,=0x%02x \\t=0x%02x\n",'\r','\n','|',',','\t');
    exit(ret);
}
